package com.alex.transcations;

import java.time.Duration;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 * @ClassName KafkaProducerInterceptor @Description TODO @Author zjg @Date 2021/12/615:28 @Version
 * 2.0
 */
public class KafkaProducerAndConsumerTrancations {
  public static void main(String[] args) {

    KafkaProducer<String, String> producer = buildKafkaProducer();
    KafkaConsumer<String, String> consumer = buildKafkaConsuemr("g1");

    // 初始化事务
    producer.initTransactions();
    consumer.subscribe(Arrays.asList("topic01"));
    while (true) {
      ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(1));
      if (!consumerRecords.isEmpty()) {
        Iterator<ConsumerRecord<String, String>> iterator = consumerRecords.iterator();
        Map<TopicPartition, OffsetAndMetadata> offset = new HashMap<>();
        // 开启生产者事务
        producer.beginTransaction();
        try {
          while (iterator.hasNext()) {
            ConsumerRecord<String, String> record = iterator.next();
            offset.put(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1) //
                );
            ProducerRecord<String, String> pRecord =
                new ProducerRecord("topic02", record.key(), record.value() + "topic test com.alex");
            producer.send(pRecord);
          }
          // 提交事务
          producer.sendOffsetsToTransaction(offset, "g1"); // 提交消费者的偏移量
          producer.commitTransaction();
        } catch (Exception e) {
          producer.abortTransaction();
        }
      }
    }
  }

  public static KafkaProducer<String, String> buildKafkaProducer() {
    //
    // 创建KakfaProducer
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "49.233.172.9:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // 必须配置事务ID，必须唯一的
    props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "www" + UUID.randomUUID().toString());
    // 配置Kafka批处理大小
    props.put(ProducerConfig.BATCH_SIZE_CONFIG, 1024);
    // 等待5ms，如果batch中的数据不足1024
    props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
    // 配置kafka的重试机制和acks
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    props.put(ProducerConfig.ACKS_CONFIG, "all");
    props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 20000);

    return new KafkaProducer<String, String>(props);
  }

  public static KafkaConsumer<String, String> buildKafkaConsuemr(String groupId) {
    // 创建KafkaConsumer
    Properties props = new Properties();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "49.233.172.9:9092");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
    props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

    // 必须关闭消费者端的offset的自动提交  需要等待消费者的业务处理完以后提交
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    return new KafkaConsumer<String, String>(props);
  }
}
